package com.doitedu.core2

import ch.hsr.geohash.GeoHash
import com.alibaba.fastjson.JSON
import com.doitedu.utils.SparkUtil
import org.apache.http.client.methods.HttpGet
import org.apache.http.impl.client.{CloseableHttpClient, HttpClients}
import org.apache.http.util.EntityUtils

import java.sql.DriverManager
/**
 * @Date: 22.7.4 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C03_补全地理位置信息 {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSparkContext("补全地理位置信息")

    // 1 加载数据
    val dataRDD = sc.textFile("data/broad/")
    // 2 解析数据   获取经纬度
    val data = dataRDD.map(line => {
      val arr = line.split(",")
      // id      latitude       longitude
      (arr(0), arr(1).toDouble, arr(2).toDouble)
    })

    //  获取mysql地理位置信息 所有的数据  JdbcRDD
    //  collect 本地  -->  (str,addess)  Map
    //  将地理位置的字典数据广播


    val geoData = data.map(tp => {
     //  (10 , 123.4324.45.34565)->str
      // 先从mysql中
      // Map.get(str)


      /**
       *  使用httpClient进行网络请求  调用的免费的API
       *     获取请求数据
       */

      // 3 根据经纬度拼接url   发送url请求
      val url = s"https://restapi.amap.com/v3/geocode/regeo?output=json&location=${tp._3},${tp._2}&key=220bd207a5b658b4ee04dceebe127738&radius=3000&extensions=bash"
      // 创建一个请求的URL对象
      val get = new HttpGet(url)
      val httpClient = HttpClients.createDefault
      // 发送请求
      val response = httpClient.execute(get)
      // 获取返回对象  获取数据
      val entity = response.getEntity
      // 解析数据
      val conent = EntityUtils.toString(entity, "utf-8")
      // 将字符串解析成JSON工具内部的一个封装数据的结构Bean
      val jsonObject = JSON.parseObject(conent)
      val address = jsonObject.getJSONObject("regeocode").getJSONObject("addressComponent")

      /**
       * 获取省市区的信息
       */

      val province = address.getString("province")
      val city = address.getString("city")
      val district = address.getString("district")
      (GeoHash.withCharacterPrecision(tp._2, tp._3, 7).toBase32, tp._1, tp._2, tp._3, province, city, district)
      // 获取属性信息
      /*      val sta = jsonObject.getString("status")
            println(sta)*/

      // Client.send(get)  --> responseContent
    })

    // 将从网络获取的数据存到MySQL
    geoData.foreachPartition(iter=>{
      // 获取连接
      val conn = DriverManager.getConnection("", "", "")
      conn.prepareStatement("insert  into  tb_district  values(? ,? , ? ,? ,? ,? ,?)")

      iter.map(tp=>{
        // 写数据到MySQL

      })
    })









    //6 将结果保存到本地


    sc.stop()
  }

}
